package org.eclipse.moquette.parser.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException;
import io.netty.util.AttributeMap;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.proto.messages.PublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * 对publish 进行解码
 * Created by alex on 15-5-25.
 */
public class PublishDecoder extends DemuxDecoder {

    private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class);

    @Override
    void decode(AttributeMap ctx, ByteBuf in, List<Object> out) throws Exception {
        LOG.info("decode invoked with buffer {}",in);

        in.resetReaderIndex();
        int startPos = in.readerIndex();

        PublishMessage message = new PublishMessage();
        if (!decodeCommonHeader(message,in)){
            LOG.info("decode ask for more data after {}",in);
            in.resetReaderIndex();
            return;
        }

        if (Utils.isMQTT3_1_1(ctx)){

            if (message.getQos() == AbstractMessage.QOSType.MOST_ONE && message.isDupFlag()) {
                throw new CorruptedFrameException("Received a PUBLISH with qos=0 && DUP = 1,MQTT 3.1.1 violation");
            }

            if (message.getQos() == AbstractMessage.QOSType.RESERVED){
                throw new CorruptedFrameException("Received a PUBLISH with QoS flags setted 10 b11, MQTT 3.1.1 violation");
            }
        }

        int remainingLength = message.getRemainingLength();

        //Topic Name
        String topic = Utils.decodeString(in);
        if (topic == null){
            in.resetReaderIndex();
            return;
        }
        if (topic.contains("*") || topic.contains("#")){
            throw new CorruptedFrameException("Received a PUBLISH with topic containting wild card chars, topic: " + topic);
        }

        message.setTopicName(topic);

        if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE ||
                message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) {
            message.setMessageID(in.readUnsignedShort());
        }

        int stopPos = in.readerIndex();

        //read the payload
        // stopPos -> 减去0 -> 减去两个固定的字节 TODO 这块要好好debug一下
        int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1);
        if (in.readableBytes() < payloadSize){
            in.resetReaderIndex();
            return;
        }

        ByteBuf bb = Unpooled.buffer(payloadSize);
        in.readBytes(bb);
        message.setPayload(bb.nioBuffer());

        out.add(message);
    }
}
